package com.atguigu.gmall.realtime.app.func;

import com.alibaba.fastjson.JSONObject;
import com.atguigu.gmall.realtime.beans.TableProcess;
import com.atguigu.gmall.realtime.common.GmallConfig;
import com.atguigu.gmall.realtime.utils.DimUtil;
import com.atguigu.gmall.realtime.utils.HbaseUtil;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;
import org.apache.flink.streaming.api.functions.sink.SinkFunction;
import org.apache.hadoop.hbase.client.Connection;

/**
 * Created by 黄凯 on 2023/7/5 0005 16:26
 *
 * @author 黄凯
 * 永远相信美好的事情总会发生.
 * <p>
 * 将流中的数据写到hbase表中
 */
public class DimSinkFunction extends RichSinkFunction<JSONObject> {

    private Connection conn;

    @Override
    public void open(Configuration parameters) throws Exception {
        conn = HbaseUtil.getConnection();
    }

    @Override
    public void close() throws Exception {
        HbaseUtil.closeConnection(conn);
    }

    @Override
    public void invoke(JSONObject jsonObj, Context context) throws Exception {

        //{"tm_name":"Redmi11","id":1,"type":"update",
        // "table_process":{"op":"r","sinkColumns":"id,tm_name","sinkFamily":"info",
        // "sinkRowKey":"id","sinkTable":"dim_base_trademark","sourceTable":"base_trademark"}}
        //获取对业务数据表数据进行的操作类型
        String type = jsonObj.getString("type");

        //获取当前维度对应的配置信息
        TableProcess tableProcess = jsonObj.getObject("table_process", TableProcess.class);

        //获取Hbase中维度表名
        String sinkTable = tableProcess.getSinkTable();

        //获取RowKey
        String rowKeyName = tableProcess.getSinkRowKey();
        String rowKey = jsonObj.getString(rowKeyName);

        //判断是否对维度表做了删除或者是修改操作
        if ("delete".equals(type) || "update".equals(type)) {

            DimUtil.delCached(sinkTable, rowKey);

        }

        if ("delete".equals(type)) {

            //如果对业务数据库维度表做了删除操作，那么需要将维度数据从Hbase中也删除掉
            System.out.println("从Hbase的" + sinkTable + "表中删除数据");
            HbaseUtil.delRow(conn, GmallConfig.HBASE_NAMESPACE, sinkTable, rowKey);

        } else {

            //如果对业务数据库维度表做了insert或者update操作，那么需要将维度数据put到Hbase中
            System.out.println("向Hbase表" + sinkTable + "中写入数据");
            //获取列族
            String sinkFamily = tableProcess.getSinkFamily();

            //获取所有列
            String[] columns = tableProcess.getSinkColumns().split(",");
            String[] values = new String[columns.length];

            for (int i = 0; i < columns.length; i++) {

                values[i] = jsonObj.getString(columns[i]);

            }

            HbaseUtil.putRow(conn, GmallConfig.HBASE_NAMESPACE, sinkTable, rowKey, sinkFamily, columns, values);


        }


    }
}
